package com.air.antispider.stream.dataprocess.businessprocess

import com.air.antispider.stream.commom.util.jedis.{JedisConnectionUtil, PropertiesUtil}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.json4s.DefaultFormats
import org.json4s.jackson.Json
import redis.clients.jedis.JedisCluster

/**
  * 链路统计功能
  */
object LinkCount {
  def linkCount(messageRDD: RDD[String]): collection.Map[String, Int] = {
    //当前一个批次每台服务器数据处理量
    val serverCountRDD: RDD[(String, Int)] = messageRDD.map(message => {
      val arr: Array[String] = message.split("#CS#")
      if (arr.length > 9) {
        //有服务端IP
        val serverIP: String = arr(9)
        if (StringUtils.isNotBlank(serverIP)) {
          //不为空，有数据
          (serverIP, 1)
        } else {
          ("", 1)
        }
      } else {
        ("", 1)
      }
    }).reduceByKey(_ + _)

    //从消息中提取当前活跃数
    val serverActiveRDD: RDD[(String, Int)] = messageRDD.map(message => {
      //切割元素
      val arr: Array[String] = message.split("#CS#")
      if (arr.length > 11) {
        //有服务端IP
        val serverIP: String = arr(9)
        val activeNum: Int = arr(11).toInt
        if (StringUtils.isNotBlank(serverIP)) {
          //如果不为空,有数据
          (serverIP, activeNum)
        } else {
          ("", 1)
        }
      } else {
        //如果取不出来数据,直接返回空ip
        ("", 1)
      }
    }).reduceByKey((x, y) => y)


    //交给前端展示
    if (!serverCountRDD.isEmpty() && !serverActiveRDD.isEmpty()) {
      //如果有数据存入redis
      //将计算结果的RDD转换为Map对象
      val serversCountMap: collection.Map[String, Int] = serverCountRDD.collectAsMap()
      val activeNumMap: collection.Map[String, Int] = serverActiveRDD.collectAsMap()
      var map: Map[String, collection.Map[String, Int]] = Map[String, collection.Map[String, Int]]()
      map += ("serversCountMap" -> serversCountMap)
      map += ("activeNumMap" -> activeNumMap)
      //将map转换为json
      val value: String = Json(DefaultFormats).write(map)

      //获取Redis连接
      val jedis: JedisCluster = JedisConnectionUtil.getJedisCluster
      //从配置文件中加载key  CSANTI_MONITOR_LP + 当前时间戳
      val key: String = PropertiesUtil.getStringByKey("cluster.key.monitor.linkProcess", "jedisConfig.properties") + System.currentTimeMillis()

      //存入Redis的延迟时间
      val ex = PropertiesUtil.getStringByKey("cluster.exptime.monitor", "jedisConfig.properties").toInt

      //保存到redis
      //      jedis.set(key,value)
      jedis.setex(key, ex, value)

    }

    //将数据处理量的结果转为Map，用作spark的任务监控
    serverCountRDD.collectAsMap()

  }
}
